package org.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/*
 window:一次对几个采集周期的数据进行处理

 */
object SparkStreaming_07 {
  def main(args: Array[String]): Unit = {
     //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(3))//Seconds(3) 采集周期
    ssc.sparkContext.setLogLevel("ERROR")

    //监听本地套接字，端口号 9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val flatMap: DStream[String] = lines.flatMap(_.split(" "))
    val map: DStream[(String, Int)] = flatMap.map((_, 1))

    /*
        窗口的范围必须是采集周期的整数倍
        第一个参数：窗口时长 （一次处理几个采集周期的数据）
        第二个参数：滑块步长 （窗口移动几个采集周期）
     */
    val winDS: DStream[(String, Int)] = map.window(Seconds(6), Seconds(6))

    val res: DStream[(String, Int)] = winDS.reduceByKey(_ + _)


    res.print()

    ssc.start()
    ssc.awaitTermination()


  }
}
